#pragma once

#include <memory>
#include <unordered_map>

#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpClient.h>

#include "detail.hpp"
#include "abstract.hpp"
#include "message.hpp"

namespace JsonRpc
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;

        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {}
        virtual ~MuduoBuffer(){}

        virtual size_t readableSize() override { return _buf->readableBytes(); }
        virtual int32_t peekInt32() override { return _buf->peekInt32(); }
        virtual void retrieveInt32() override { return _buf->retrieveInt32(); }
        virtual int32_t readInt32() override { return _buf->readInt32(); }
        virtual std::string retrieveAsString(size_t len) override { return _buf->retrieveAsString(len); }

    protected:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template<typename ...Args>
        static BaseBuffer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol //应用层协议
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;

        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        virtual ~LVProtocol(){}
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if (buf->readableSize() < lenFieldsLength)
                return false;
            int32_t total_len = buf->peekInt32();
            if (total_len + lenFieldsLength > buf->readableSize())
                return false;
            return true;
        }

        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            //默认缓冲区中至少有一条完整的数据
            int32_t total_len = buf->readInt32();
            MType mtype = static_cast<MType>(buf->readInt32());

            int32_t idlen = buf->readInt32();
            std::string id = buf->retrieveAsString(idlen);

            size_t body_len = total_len - mtypeFieldsLength - idlenFieldsLength - idlen;
            std::string body = buf->retrieveAsString(body_len);  
            

            msg = MessageFactory::create(mtype);
            if (msg == nullptr)
            {
                LOG(ERROR, "消息类型错误，构造消息对象失败！\n");
                return false;
            }
            if (!msg->deserialize(body))
            {
                LOG(ERROR, "正文反序列化失败！\n");
                return false;
            }
            msg->setMType(mtype);
            msg->setId(id);
            return true;
        }

        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            // |--Len--|--mtype--|--idlen--|--id--|--body--|
            std::string body = msg->serialize();
            std::string id = msg->rid();
            auto mtype = htonl(static_cast<int32_t>(msg->Mtype()));
            int32_t idlen = htonl(id.size());
            int32_t h_total_len = mtypeFieldsLength + idlenFieldsLength + idlen + body.size();
            int32_t n_total_len = htonl(h_total_len);

            std::string result;
            result.reserve(h_total_len);
            result.append(reinterpret_cast<const char*>(&n_total_len, lenFieldsLength));
            result.append(reinterpret_cast<const char*>(&mtype, mtypeFieldsLength));
            result.append(reinterpret_cast<const char*>(&idlen, idlenFieldsLength));
            result.append(id);
            result.append(body);
            return result;
        }

    protected:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idlenFieldsLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template<typename ...Args>
        static BaseProtocol::ptr create(Args&& ...args)
        {
            return std::make_shared<BaseProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;

        MuduoConnection(BaseProtocol::ptr &protocol, muduo::net::TcpConnectionPtr &conn)
        :_protocol(protocol), _conn(conn)
        {}
        virtual ~MuduoConnection(){}

        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string body = _protocol->serialize(msg);
            _conn->send(body);
        }
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        virtual bool connected() override
        {
            return _conn->connected();
        }
    protected:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template<typename ...Args>
        static BaseConnection::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        
        MuduoServer(uint16_t port)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort)
        ,_protocol(ProtocolFactory::create())
        {}
        virtual ~MuduoServer() {}

        void start()
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::OnConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::OnMessage, this,
                                       std::placeholders::_1, std::placeholders::_2,
                                       std::placeholders::_3));
            _server.start();
            _baseloop.loop();
        }

    private:
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                std::cout << "连接建立" << std::endl;
                BaseConnection::ptr muduo_conn = ConnectionFactory::create(_protocol, conn);

                std::unique_lock<std::mutex> lock(_mutex);
                _conns.insert(std::make_pair(conn, muduo_conn));

                if (_cb_connection)
                {
                    _cb_connection(muduo_conn);
                }
            }
            else
            {
                std::cout << "连接关闭" << std::endl;
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _conns.find(conn);
                if (it == _conns.end())
                {
                    return;
                }
                BaseConnection::ptr muduo_conn = _conns[conn];
                _conns.erase(conn);

                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }

        void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG(DEBUG, "连接有数据到来，开始处理！\n");
            
            auto it = _conns.find(conn);
            if (it == _conns.end())
            {
                conn->shutdown();
                LOG(WARNING, "连接丢失！\n");
            }

            BaseBuffer::ptr base_buf = BufferFactory::create(buf);
            while (true)
            {
                if (!_protocol->canProcessed(base_buf))
                {
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        std::unique_lock<std::mutex> lock(_mutex);
                        _conns.erase(conn);
                        LOG(WARNING, "缓冲区数据过大\n");
                    }
                    //数据不足
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false)
                {
                    conn->shutdown();
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.erase(conn);
                    LOG(WARNING, "缓冲区数据错误!\n");
                }
                if (_cb_message)
                    _cb_message(_conns[conn], msg);
            }
        }
    protected:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };

    class ServerFactory
    {
    public:
        template<typename ...Args>
        static BaseServer::ptr create(Args&& ...args)
        {
            return make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
            
        MuduoClient(const std::string &sip, const uint16_t sport)
        :_baseloop(_loopthread.startLoop()),
        _client(_baseloop, muduo::net::InetAddress(sip, sport), "DictClient"),
        _downlatch(1)
        {
            //设置connection回调
            _client.setConnectionCallback(std::bind(&MuduoClient::OnConnection, this, std::placeholders::_1));
            //设置消息回调
            _client.setMessageCallback(std::bind(&MuduoClient::OnMessage, this,
                std::placeholders::_1,
                std::placeholders::_2,
                std::placeholders::_3));
            
            //与服务器建立连接
            _client.connect();
            //进程同步
            _downlatch.wait();
        }
        virtual ~MuduoClient() {}
        

    private:
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                std::cout << "连接建立" << std::endl;
                BaseConnection::ptr muduo_conn = ConnectionFactory::create(_protocol, conn);

                std::unique_lock<std::mutex> lock(_mutex);
                _conns.insert(std::make_pair(conn, muduo_conn));

                if (_cb_connection)
                {
                    _cb_connection(muduo_conn);
                }
            }
            else
            {
                std::cout << "连接关闭" << std::endl;
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _conns.find(conn);
                if (it == _conns.end())
                {
                    return;
                }
                BaseConnection::ptr muduo_conn = _conns[conn];
                _conns.erase(conn);

                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }

        void OnMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            LOG(DEBUG, "连接有数据到来，开始处理！\n");
            
            auto it = _conns.find(conn);
            if (it == _conns.end())
            {
                conn->shutdown();
                LOG(WARNING, "连接丢失！\n");
            }

            BaseBuffer::ptr base_buf = BufferFactory::create(buf);
            while (true)
            {
                if (!_protocol->canProcessed(base_buf))
                {
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        std::unique_lock<std::mutex> lock(_mutex);
                        _conns.erase(conn);
                        LOG(WARNING, "缓冲区数据过大\n");
                    }
                    //数据不足
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false)
                {
                    conn->shutdown();
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.erase(conn);
                    LOG(WARNING, "缓冲区数据错误!\n");
                }
                if (_cb_message)
                    _cb_message(_conns[conn], msg);
            }
        }

    protected:
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::CountDownLatch _downlatch;
        BaseConnection::ptr _conn;
        muduo::net::TcpClient _client;
    };
}
